package client

import (
	"encoding/json"
	"fmt"
	"gark-elastic-job/common"
	"gark-elastic-job/util"
	"sync"
	"time"

	logger "github.com/sirupsen/logrus"
)

type SnapshotProcessor struct {
	jobs             map[string]Job
	executionLogPath string
	snapshotPath     string
	snapshotsChan    chan *common.GarkJobSnapshot

	locker *sync.RWMutex
	etcd   *Etcd
}

func NewSnapshotProcessor(id, group string, etcd *Etcd) *SnapshotProcessor {
	sp := &SnapshotProcessor{
		snapshotsChan:    make(chan *common.GarkJobSnapshot, 512),
		etcd:             etcd,
		jobs:             make(map[string]Job),
		executionLogPath: fmt.Sprintf(jobExecuteLogPath, group, id),
		snapshotPath:     fmt.Sprintf(jobSnapshotPath, group, id),
		locker:           &sync.RWMutex{},
	}
	go sp.lookup()
	return sp
}

func (p *SnapshotProcessor) pushSnapshot(snapshot *common.GarkJobSnapshot) {
	p.snapshotsChan <- snapshot
}

func (p *SnapshotProcessor) lookup() {
	for {
		select {
		case snapshot := <-p.snapshotsChan:
			p.handleSnapshot(snapshot)
		}
	}
}

func (p *SnapshotProcessor) handleSnapshot(snapshot *common.GarkJobSnapshot) {

	target := snapshot.Target
	now := time.Now()

	jobExecutionLog := common.JobExecutionLog{
		ID:         util.GenerateId(),
		JobId:      snapshot.Id,
		Name:       snapshot.Name,
		Ip:         snapshot.IP,
		Group:      snapshot.Group,
		Cron:       snapshot.Cron,
		Target:     target,
		Payload:    snapshot.Payload,
		Remark:     snapshot.Remark,
		CreateTime: snapshot.CreateTime,
		Status:     common.Running,
		StartTime:  util.Format(now),
	}

	if target == "" {
		logger.Errorf("job %s execution target can't be empty", snapshot.Name)
		jobExecutionLog.Status = common.Unknown
		return
	}

	job, ok := p.jobs[target]

	if !ok || job == nil {
		logger.Infof("the snapshot: %v target not found in this job list", snapshot)
		jobExecutionLog.Status = common.Unknown
		time.Sleep(3 * time.Second)
		return
	}

	value, err := json.Marshal(jobExecutionLog)
	if err != nil {
		logger.Errorf("encode jobExecutionLog:%v failed, err: %v", jobExecutionLog, err)
		jobExecutionLog.Status = common.Unknown
	}

	key := p.executionLogPath + jobExecutionLog.ID
	if err = p.etcd.Put(key, string(value)); err != nil {
		logger.Errorf("failed to put executionLog:%v to the path: %s", jobExecutionLog, key)
		return
	}

	if jobExecutionLog.Status != common.Running {
		return
	}
	result, err := job.Execute(snapshot.Payload)
	// 执行结果判断
	if err != nil {
		jobExecutionLog.Status = common.Failed
		jobExecutionLog.Result = err.Error()
	} else {
		jobExecutionLog.Status = common.Success
		jobExecutionLog.Result = result
	}
	// 计算耗时
	endTime := time.Now()
	times := endTime.Sub(now) / time.Second
	jobExecutionLog.Times = int(times)
	jobExecutionLog.FinishTime = util.Format(endTime)

	logger.Infof("the execute snapshot:%v execute success ", jobExecutionLog)

	// collect the execute job snapshot
	value, _ = json.Marshal(jobExecutionLog)
	if err := p.etcd.Put(key, string(value)); err != nil {
		logger.Warnf("the snapshot:%v put update snapshot execute fail:%v", jobExecutionLog, err)
		return
	}
}

func (p *SnapshotProcessor) PushJob(name string, job Job) {
	p.locker.Lock()
	defer p.locker.Unlock()

	if _, ok := p.jobs[name]; ok {
		logger.Infof("job %s: %v already exists", name, job)
		return
	}

	p.jobs[name] = job
}
